package com.linx.rxlife

import androidx.lifecycle.Lifecycle
import androidx.lifecycle.LifecycleOwner
import io.reactivex.CompletableSource
import io.reactivex.MaybeSource
import io.reactivex.ObservableSource
import io.reactivex.SingleSource
import io.reactivex.disposables.Disposable
import io.reactivex.functions.Action
import io.reactivex.functions.Consumer
import io.reactivex.internal.functions.Functions
import org.reactivestreams.Publisher
import org.reactivestreams.Subscription

fun <T> Publisher<T>.observe(
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED,
    onError: ((Throwable) -> Unit)? = null,
    onComplete: (() -> Unit)? = null,
    onSubscribe: ((Subscription) -> Unit)? = null,
    onNext: (T) -> Unit
): Disposable {
    val life = FlowableLife(this, owner, atLeastState)
    life.observe(
        Consumer { onNext(it) },
        onError?.let { error -> Consumer<Throwable> { error(it) } } ?: Functions.ON_ERROR_MISSING,
        onComplete?.let { complete -> Action { complete() } } ?: Functions.EMPTY_ACTION,
        onSubscribe?.let { Consumer<Subscription> { subscription -> it(subscription) } } ?: Functions.emptyConsumer()
    )
    return life
}

fun <T> ObservableSource<T>.observe(
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED,
    onError: ((Throwable) -> Unit)? = null,
    onComplete: (() -> Unit)? = null,
    onSubscribe: ((Disposable) -> Unit)? = null,
    onNext: (T) -> Unit
): Disposable {
    val life = ObservableLife(this, owner, atLeastState)
    life.observe(
        Consumer { onNext(it) },
        onError?.let { error -> Consumer<Throwable> { error(it) } } ?: Functions.ON_ERROR_MISSING,
        onComplete?.let { complete -> Action { complete() } } ?: Functions.EMPTY_ACTION,
        onSubscribe?.let { Consumer<Disposable> { disposable -> it(disposable) } } ?: Functions.emptyConsumer()
    )
    return life
}

fun <T> SingleSource<T>.observe(
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED,
    onError: ((Throwable) -> Unit)? = null,
    onSuccess: (T) -> Unit
): Disposable {
    val life = SingleLife(this, owner, atLeastState)
    life.observe(
        Consumer { onSuccess(it) },
        onError?.let { error -> Consumer<Throwable> { error(it) } } ?: Functions.ON_ERROR_MISSING
    )
    return life
}

fun <T> MaybeSource<T>.observe(
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED,
    onError: ((Throwable) -> Unit)? = null,
    onComplete: (() -> Unit)? = null,
    onSuccess: (T) -> Unit
): Disposable {
    val life = MaybeLife(this, owner, atLeastState)
    life.observe(
        Consumer { onSuccess(it) },
        onError?.let { error -> Consumer<Throwable> { error(it) } } ?: Functions.ON_ERROR_MISSING,
        onComplete?.let { complete -> Action { complete() } } ?: Functions.EMPTY_ACTION
    )
    return life
}

fun CompletableSource.observe(
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED,
    onError: ((Throwable) -> Unit)? = null,
    onComplete: (() -> Unit)? = null
): Disposable {
    val life = CompletableLife(this, owner, atLeastState)
    life.observe(
        onError?.let { error -> Consumer<Throwable> { error(it) } } ?: Functions.ON_ERROR_MISSING,
        onComplete?.let { complete -> Action { complete() } } ?: Functions.EMPTY_ACTION
    )
    return life
}